package com.example.mqtt.base;

import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;

import java.util.List;
import java.util.stream.IntStream;

@Slf4j
@AllArgsConstructor
public class MqttCallback implements MqttCallbackExtended {
    private final MqttClient mqttClient;
    private final List<? extends MqttReceiver> mqttReceivers;

    @Override
    public void connectComplete(boolean reconnect, String serverUrl) {
        log.info("MQTT连接成功：clientId=" + mqttClient.getClientId());
        try {
            String[] topicFilters = mqttReceivers.stream()
                    .map(MqttReceiver::subTopic)
                    .map(MqttSubTopic::getShareTopic)
                    .toArray(String[]::new);
            int[] qos = mqttReceivers.stream()
                    .map(MqttReceiver::subTopic)
                    .mapToInt(MqttSubTopic::getQos)
                    .toArray();
            mqttClient.subscribeWithResponse(topicFilters, qos).setActionCallback(new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    String[] topics = asyncActionToken.getTopics();
                    int[] grantedQos = asyncActionToken.getGrantedQos();
                    IntStream.range(0, topics.length).forEach(i -> log.info("MQTT订阅成功：qos=" + grantedQos[i] + " topic=" + topics[i]));
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable e) {
                    String[] topics = asyncActionToken.getTopics();
                    int[] grantedQos = asyncActionToken.getGrantedQos();
                    IntStream.range(0, topics.length).forEach(i -> log.info("MQTT订阅失败：qos=" + grantedQos[i] + " topic=" + topics[i]));
                }
            });
        } catch (Throwable e) {
            //在订阅时发生异常会触发connectionLost方法
            throw new RuntimeException("MQTT订阅异常：", e);
        }
    }

    @Override
    @SneakyThrows
    public void connectionLost(Throwable e) {
        log.error("MQTT连接断开", e);
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        for (MqttReceiver listener : mqttReceivers) {
            if (MqttTopic.isMatched(listener.subTopic().getTopic(), topic)) {
                listener.messageArrived(topic, message);
            }
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
    }

}
